package com.baidu.noah.zeus.master;

import java.io.File;
import java.io.IOException;
import java.net.UnknownHostException;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Server;

import com.baidu.noah.zeus.config.ZConstants;
import com.baidu.noah.zeus.protocol.InterTrackerProtocol;
import com.baidu.noah.zeus.vo.HeartbeatResponse;
import com.baidu.noah.zeus.vo.TaskProgressInfo;

public class JobTracker implements Runnable, InterTrackerProtocol {
	
	private Server server;

	@Override
	public void run() {
		try {
			server = RPC
					.getServer(this, "localhost", 8888, new Configuration());
			server.start();
			server.join();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	@Override
	public long getProtocolVersion(String arg0, long arg1) throws IOException {
		return 1;
	}

	@Override
	public HeartbeatResponse heartbeat() {
		try {
			File tasksDir = new File("/tmp/zeus/tasks/todo");
			if (tasksDir.exists() && tasksDir.canRead() && tasksDir.isDirectory()) {
				String[] files = FileUtil.list(tasksDir);
				if(files == null || files.length == 0) {
					return new HeartbeatResponse(false, -1, "");
				} else {
					String filename = files[0];
					File file = new File("/tmp/zeus/tasks/todo/" + filename);
					HeartbeatResponse ret =  new HeartbeatResponse(true, Integer.parseInt(filename), FileUtils.readFileToString(file));
					FileUtils.moveFile(file, new File("/tmp/zeus/tasks/doing/" + filename));
					return ret;
				}
			} else {
				return new HeartbeatResponse(false, -1, "");
			}
		} catch (IOException e) {
			e.printStackTrace();
			return new HeartbeatResponse(false, -1, "");
		}
	}

	@Override
	public Text pln() {
		String tmp = Math.round(Math.random() * 100) + "";
		System.out.println(tmp);
		return new Text(tmp);
	}

	@Override
	public boolean reportTaskProgress(TaskProgressInfo info) {
		long taskId = info.getTaskId();
		// 此处之后可以改成设计模式
		if (info.getTaskStatus() == ZConstants.TASK_STATUS_COMPLETE) {
			try {
				File src = new File("/tmp/zeus/tasks/doing/" + taskId);
				if (src.exists()) {
					FileUtils.moveFile(src, new File("/tmp/zeus/tasks/did/" + taskId));
				}
			} catch (IOException e) {
				e.printStackTrace();
				// log
				return false;
			}
		} else if (info.getTaskStatus() == ZConstants.TASK_STATUS_DOING) {
			
		}
			
		return true;
	}

}
